package com.yn.xgame.gate;

import com.google.common.collect.ImmutableMap;
import com.yn.xgame.*;
import com.yn.xgame.message.C2GPbMessage;
import com.yn.xgame.pb.ClientGateProto;
import com.yn.xgame.po.LoadInfo;
import com.yn.xgame.util.CtxUtils;
import io.netty.channel.ChannelHandlerContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * GateApp
 * Created by wangboo on 17/7/12.
 */
public class GateApp implements NettyProtobufHandler.Callback {

    private static final Logger logger = LoggerFactory.getLogger(GateApp.class);
    static GateServer gateServer;
    static NettyServer nettyServer;
    static CtxManager ctxManager;
    static ZkClient zkClient;
    static LoadInfoWatcher loadInfoWatcher;
    static CommonConfig config;
    static MQClient mqClient;

    public static void main(String[] args) throws Exception {
        GateApp instance = new GateApp();
        String file = "gate1.cfg";
        if (args.length == 1) {
            file = args[0];
        }
        config = new CommonConfig("gate", file);
        gateServer = new GateServer();
        nettyServer = new NettyServer(config.getPort(), instance);
        ctxManager = nettyServer.getCtxManager();
        zkClient = new ZkClient();
        loadInfoWatcher = new LoadInfoWatcher(zkClient, GlobalSetting.ZK_PREFIX + "/logic",
                gateServer.loadInfoManager, gateServer);
        CommonMqMessageHandler mqHandler = new CommonMqMessageHandler(gateServer);
        mqHandler.setCtxManager(ctxManager);
        mqClient = new MQClient(config.getKafkaProp(), mqHandler, config.getMqGroup(),
                config.getTopic());
        mqClient.start();
        gateServer.startup();
        zkClient.connect(config.getZkAddress(), loadInfoWatcher);
        // 寻找logic服务器
        zkClient.getChildren("/logic").forEach(id -> {
            LoadInfo info = zkClient.getData("/logic/" + id, LoadInfo.class);
            gateServer.loadInfoManager.addOrUpdate(info);
        });
        // 向zk注册自己节点
        zkClient.createOrsetData_EPHEMERAL("/gate/" + config.getId(), gateServer.loadInfo);
        loadInfoWatcher.doWatch();
        nettyServer.start();
    }

    @Override
    public void handle(ChannelHandlerContext ctx, byte[] bin) {
        try {
            ClientGateProto.C2GMessage msg = ClientGateProto.C2GMessage.parseFrom(bin);
            C2GPbMessage pbMsg = new C2GPbMessage(ctx, msg);
            gateServer.pushMessage(pbMsg);
        } catch (Exception e) {
            logger.error("decode error: ", e);
        }
    }

    @Override
    public void channelUnregistered(ChannelHandlerContext ctx) {
        String logicTopic = CtxUtils.getLogicTopic(ctx);
        if (logicTopic == null) return;
        long sid = CtxUtils.getId(ctx);
        // 通知logic，玩家掉线
        mqClient.sendInter(config.getTopic(), logicTopic, MqMsgId.gate_user_disconnect, ImmutableMap.of("sid", sid));
    }

}
